package com.tyylab.message.core.service;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.tyylab.message.core.config.properties.TyylabMessageRedisProperties;
import com.tyylab.message.core.intf.IConditionalSubscribe;
import com.tyylab.message.core.intf.ISubscribe;
import com.tyylab.message.core.model.ProcessingMessage;
import com.tyylab.message.core.model.TransMessage;
import io.lettuce.core.KeyValue;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.env.Environment;

import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

@Slf4j
public class ConcurrentSubscribeService implements ISubscribeSupport {
    @Autowired
    @Qualifier("readBusConn")
    StatefulRedisConnection<String,String> redisConnection;

    @Autowired
    TyylabMessageRedisProperties simulMessageRedisProperties;

    @Autowired
    ApplicationContext ctx;

    Boolean canSubscribe = false;

    int blockTimeout = 100; //s

    Map<String, Set<ISubscribe>> subscribes = Maps.newLinkedHashMap();
    Map<String, Set<String>> subscribesClz = Maps.newLinkedHashMap();

    Set<IConditionalSubscribe> conditionalSubscribes = Sets.newConcurrentHashSet();
    Set<String> conditionalSubscribesClz = Sets.newConcurrentHashSet();

    Map<String, Object> nameSubscribeMap = Maps.newHashMap();
    Map<String, Object> nameConditionalSubscribeMap = Maps.newHashMap();

    String ip ="";
    String app="";

    private void init(){
        wireSubscribes();
        initDelStatefulConnection();
        CompletableFuture.runAsync(()->{
            for (long i = 1; i < 11L; i++) {
                try{
                    Thread.sleep(1000* i);
                    wireSubscribes();
                }
                catch (Exception e){
                    log.error(e.getMessage(),e);
                }
            }

        });

        try {
            Environment environment = (Environment)this.ctx.getBean(Environment.class);
            this.app = environment.getProperty("spring.application.name");
            InetAddress address = InetAddress.getLocalHost();
            this.ip = address.getHostAddress();
        } catch (UnknownHostException var3) {
            var3.printStackTrace();
        }
    }

    @EventListener(ContextRefreshedEvent.class)
    public void start(ContextRefreshedEvent e){
        log.info("开启消息监听..");
        onSubscribe();
        canSubscribe = true;
    }

    private void wireSubscribes() {
        Map<String, ISubscribe> beans = ctx.getBeansOfType(ISubscribe.class);

        beans.values()
                .stream()
                .forEach(bean->{
                    this.isSubscribeQualified(bean);
                });

        subscribes.clear();
        subscribesClz.clear();
        for (Object val
                : this.nameSubscribeMap.values()) {
            ISubscribe support = ((ISubscribe)val);
            String type = support.getSubscribeType();
            if (subscribes.containsKey(type) && !subscribes.get(type).contains(support)){
                synchronized (this) {
                    if (!subscribesClz.get(type).contains(support.getClass().getName())) {
                        subscribes.get(type).add(support);
                    }
                }
            }
            else{
                synchronized (this) {
                    Set<ISubscribe> subscribesList = Sets.newConcurrentHashSet();
                    subscribesList.add(support);
                    subscribes.put(type, subscribesList);

                    Set<String> subscribesClzList = Sets.newConcurrentHashSet();
                    subscribesClzList.add(support.getClass().getName());
                    subscribesClz.put(type, subscribesClzList);
                }
            }
        }
        // log
        log.info("========= finish reload ISubscribe =========");
        for (Set<ISubscribe> subs :
                subscribes.values()) {
            for (ISubscribe sub :
                    subs) {
                log.info("regist ISubscribe=>" + sub.getClass());
            }
        }

        Map<String, IConditionalSubscribe> conditonalBeans = ctx.getBeansOfType(IConditionalSubscribe.class);
        conditonalBeans.values()
                .stream()
                .forEach(bean->{
                    this.isConditionalSubscribeQualified(bean);
                });

        conditionalSubscribes.clear();
        conditionalSubscribesClz.clear();

        if (null!= nameConditionalSubscribeMap && !nameConditionalSubscribeMap.isEmpty()){
            nameConditionalSubscribeMap.values().stream()
                    .filter(b->{
                        return  !conditionalSubscribes.contains(b)
                                && !conditionalSubscribesClz.contains(b.getClass().getName())
                                ;
                    })
                    .forEach(b->{
//                        log.info("regist IConditionalSubscribe=>" + b.getClass());
                        conditionalSubscribes.add((IConditionalSubscribe) b);
                        conditionalSubscribesClz.add(b.getClass().getName());

                    });
        }

        // log
        log.info("========= finish reload IConditionalSubscribe =========");
        for (IConditionalSubscribe sub :
                conditionalSubscribes) {
            log.info("regist IConditionalSubscribe=>" + sub.getClass());
        }
    }

    boolean hasStarted= false;
    //@Autowired
    @Override
    public void onSubscribe() {
        if (hasStarted){
            return;
        }
        hasStarted = true;
        log.info("init..");
        init();
        final String channel = simulMessageRedisProperties.getChannel();
        log.info("start subscribe..");
        int threads = getProcessorCount() * simulMessageRedisProperties.getThreadsOfEachProcessor();
        //executor
        ThreadPoolExecutor executor = new ThreadPoolExecutor(threads + 1,
                threads + 1,
                10L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
        //
        log.info("启动 " + threads + "个监听线程...");
        for (int i = 0; i < threads; i++) {
            CompletableFuture.runAsync(()->{
                try{
                    log.info("等待1分钟，开始监听消息");
                    Thread.sleep(60*1000);
                    log.info("开始message subsribe");
                }
                catch (Exception e){
                    log.error(e.getMessage(),e);
                }

                while (true) {

                    try {
                        if (!canSubscribe){
                            log.info("未开启message subsribe....");
                            Thread.sleep(3000);
                            continue;
                        }
                        KeyValue<String,String> kv = SyncPopMessage(channel);
                        if (null != kv ) {
                            onMessage(kv.getValue());
                        } else {
                            // 结果为空等两秒
                            try {
                                Thread.sleep(2000);
                            } catch (Exception e) {
                            }
                        }
                    }
                    catch (Exception e){
                        log.error(e.getMessage(),e);
                        try {
                            Thread.sleep(2000);
                        } catch (Exception ex) {
                        }
                    }
                }


            }, executor);
        }

        log.info("启动消息失败补偿处理线程..");
        CompletableFuture.runAsync(()->{
            try{
                log.info("等待1分钟，准备处理失败消息");
                Thread.sleep(60*1000);
                log.info("开始处理失败消息");
                reprocess();
                log.info("消息失败补偿处理 完毕！");
            }
            catch (Exception e){
                log.error(e.getMessage(),e);
            }

        }, executor);

    }

    /**
     * 同步获取redis消息
     * @param channel
     * @return
     */
    KeyValue<String, String> SyncPopMessage(String channel) {
        return redisConnection.sync().brpop(blockTimeout, channel);
    }

    private void onMessage(final String msg){
        if (StringUtils.isNotBlank(msg)){
            log.info("onMessage => " + msg );
            TransMessage transMessage = JSON.parseObject(msg, TransMessage.class);
            // before
            before(transMessage);
            try {
                if (subscribes.containsKey(transMessage.getClz())) {
                    for (ISubscribe s :
                            subscribes.get(transMessage.getClz())) {
                        try {
                            s.onMesssage(transMessage);
                        } catch (Exception e) {
                            log.error(e.getMessage() + ",msg=>" + msg, e);
                        }
                    }
                }
                for (IConditionalSubscribe conditonalSubscribe :
                        conditionalSubscribes) {
                    if (conditonalSubscribe.isSubscribe(transMessage.getClz())) {
                        try {
                            conditonalSubscribe.onMesssage(transMessage);
                        } catch (Exception e) {
                            log.error(e.getMessage() + ",msg=>" + msg, e);
                        }
                    }
                }
            }
            catch (Exception ex){
                log.error(ex.getMessage(),ex);
            }
            //after
            after(transMessage);
        }
        else{
            log.warn("msg is blank.");
        }
    }

    private void before(TransMessage transMessage){
        long ts = System.currentTimeMillis();
        ProcessingMessage processingMessage = ProcessingMessage.builder()
                .message(transMessage)
                .createTime(ts)
                .app(app)
                .processIP(ip)
                .thread(Thread.currentThread().getName())
                .build();
        String key = getProcessingHKey(transMessage);
        String id = transMessage.getId();
        if (StringUtils.isBlank(transMessage.getId())){
            id = transMessage.getMessage().getString("id");
        }
        String js = JSON.toJSONString(processingMessage);
        getLogStatefulConnection().sync().hset(key,id,js);

        // 删除唯一key
        if (null!= transMessage.getIsSupportUnique() && transMessage.getIsSupportUnique() >0){
            int isUnque = transMessage.getMessage().containsKey(IS_UNIQUE_KEY_NAME)? transMessage.getMessage().getIntValue(IS_UNIQUE_KEY_NAME) : 0;
            if (isUnque>0){
                if (transMessage.getMessage().containsKey(HASH_KEY_NAME)){
                    String hashkey = transMessage.getMessage().getString(HASH_KEY_NAME);
                    if (StringUtils.isNotBlank(hashkey)) {
                        uniqueSupport.removeUniqueKey(simulMessageRedisProperties.getChannel(), hashkey +":"+ transMessage.getClz());
                        log.info("remove unique :" + hashkey);
                    }
                }
            }
        }
    }

    @Autowired
    IUniqueSupport uniqueSupport;

    static final String IS_UNIQUE_KEY_NAME = "isUnique";
    static final String HASH_KEY_NAME = "hash";

    // after

//    private void after(SimulMessage simulMessage){
//        String key = getProcessingKey(simulMessage);
//        getDelStatefulConnection().sync().del(key);
//        log.info("delete :" + key);
//    }

    private void after(TransMessage transMessage){
        String key = getProcessingHKey(transMessage);
        String id = transMessage.getId();
        if (StringUtils.isBlank(transMessage.getId())){
            id = transMessage.getMessage().getString("id");
        }

        getLogStatefulConnection().sync().hdel(key,id);
        log.info("delete :" + key +":" + id);
    }


    private void reprocess(){
        log.info("处理失败消息...");
        String keysPre = simulMessageRedisProperties.getChannel()+ ":processing" ;
        List<String> keys = hkeys(keysPre);

        log.info("共"+ (null == keys ? 0 : keys.size())+ "个失败消息：");
        if (null!= keys && !keys.isEmpty()){
            for (String key :
                    keys) {
                if (StringUtils.isNotBlank(key)) {
                    String msg = getLogStatefulConnection().sync().hget(keysPre,key);
                    if (StringUtils.isBlank(msg)){
                        getLogStatefulConnection().sync().hdel(keysPre,key);
                        continue;
                    }
                    log.info("处理=>" + msg);
                    ProcessingMessage processingMessage = JSON.parseObject(msg, ProcessingMessage.class);
                    if ((System.currentTimeMillis() - processingMessage.getCreateTime()) > simulMessageRedisProperties.getReprocessInterval() *1000) {
                        onMessage(JSON.toJSONString(processingMessage.getMessage()));
                    }
                }
            }
        }
    }


    private List<String> hkeys(String hkey){
        return redisConnection.sync().hkeys(hkey);
    }


    private String getProcessingHKey(TransMessage transMessage){
        return simulMessageRedisProperties.getChannel()+ ":processing";
    }

    //
    // 使用独立的redis通道来删除
    StatefulRedisConnection<String,String> logstatefulRedisConnection;
    StatefulRedisConnection<String,String> getLogStatefulConnection(){
        return logstatefulRedisConnection;
    }
    void initDelStatefulConnection(){
        RedisURI redisUri = RedisURI.builder()                    // <1> 创建单机连接的连接信息
                .withHost(simulMessageRedisProperties.getHost())
                .withPort(simulMessageRedisProperties.getPort())
                .withTimeout(Duration.of(simulMessageRedisProperties.getTimeout(), ChronoUnit.SECONDS))
                .withPassword(simulMessageRedisProperties.getPassword())
                .withDatabase(simulMessageRedisProperties.getDatabase())
                .build();
        RedisClient redisClient = RedisClient.create(redisUri);   // <2> 创建客户端
        StatefulRedisConnection<String, String> connection = redisClient.connect();     // <3> 创建线程安全的连接
        this.logstatefulRedisConnection = connection;
    }

    // 是否使用监听器

    private boolean isSubscribeQualified(ISubscribe subscribe){
        return isQualified(nameSubscribeMap,subscribe);
    }

    private boolean isConditionalSubscribeQualified(IConditionalSubscribe subscribe){
        return isQualified(nameConditionalSubscribeMap,subscribe);
    }

    private boolean isQualified(Map<String, Object> map, Object object){
        String oriClzName = object.getClass().getName();
        String clzName = oriClzName.contains("$") ?
                oriClzName.substring(0,oriClzName.indexOf("$"))
                : oriClzName;
        if(!map.containsKey(clzName)){
            map.put(clzName,object);
            return true;
        }
        else {
            String existName = map.get(clzName).getClass().getName();
            if (existName.length() < oriClzName.length()){
                map.put(clzName,object);
                return true;
            }
            else{
                return false;
            }
        }
    }

    static int getProcessorCount(){
        OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getPlatformMXBean(OperatingSystemMXBean.class);

        return operatingSystemMXBean.getAvailableProcessors();
    }
}
